package com.lianjia.sh.se.kafka.client;

import com.lianjia.sh.se.kafka.client.config.CustomDefaultSenderConfig;
import com.lianjia.sh.se.kafka.client.config.ClientProducerDefaultConfig;
import com.lianjia.sh.se.kafka.client.constants.ProducerCons;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * Created by yangsp1 on 16/7/28.
 */
@Component
public class KafkaProducerFactory {
    @Autowired
    ClientProducerDefaultConfig producerDefaultConfig;
    
    @Autowired
    CustomDefaultSenderConfig customDefaultSenderConfig;

    Map<AbstractKafkaSender,Producer> producerCache = new HashMap<>();

    /**
     * 创建：生产者
     * @return
     */
    public Producer createProducer(AbstractKafkaSender bean, Properties configs){
        if (this.producerCache.get(bean) == null) {
            synchronized (bean) {
                if (this.producerCache.get(bean) == null) {
                    Properties populateConfigs = populateConfigs(configs);

                    //生成key和value反序列化对象
                    String keySerializerClass = (String)populateConfigs.get(ProducerCons.KEY_SERIALIZER_NAME);
                    String valueSerializerClass = (String)populateConfigs.get(ProducerCons.VALUE_SERIALIZER_NAME);

                    Serializer keySerializer = null;
                    try {
                        keySerializer = (Serializer) Class.forName(keySerializerClass).newInstance();
                    }catch (Exception e){
                        keySerializer = new StringSerializer();
                    }

                    Serializer valueSerializer = null;
                    try {
                        valueSerializer = (Serializer) Class.forName(valueSerializerClass).newInstance();
                    }catch (Exception e){
                        valueSerializer = new StringSerializer();
                    }

                    //生成生产者并缓存
                    this.producerCache.put(bean,new KafkaProducer(populateConfigs, keySerializer, valueSerializer));
                }
            }
        }
        return this.producerCache.get(bean);
    }


    /**
     * 计算消费都配置
     * @param customConfigs
     * @return
     */
    private Properties populateConfigs(Properties customConfigs){
        //首先使用listener自定义配置
        Properties config = populateCustomConfigs(customConfigs);

        //其次使用应用统一自定义默认配置
        config = populateAppDefaultCustomConfigs(config);

        //全局统一默认配置
        config = populateCommonDefaultConfigs(config);

        //最后：使用kafak原始的默认值
        return config;
    }


    /**
     * listener自身的自定义配置
     * @return
     */
    private Properties populateCustomConfigs(Properties customConfigs){
        Properties rs = new Properties();
        rs.putAll(customConfigs);
        return rs;
    }

    /**
     * 默认的自定义配置（应用级)
     * @return
     */
    private Properties populateAppDefaultCustomConfigs(Properties configs){
        Properties rs = new Properties();
        rs.putAll(configs);
        //broker 地址列表
        String bootstrapServers = (String)configs.get(ProducerCons.BOOTSTRAP_SERVERS_NAME);
        String bootstrapServersCustom = customDefaultSenderConfig.getBootstrapServers();
        if(StringUtils.isBlank(bootstrapServers) && StringUtils.isNotBlank(bootstrapServersCustom)){
            rs.put(ProducerCons.BOOTSTRAP_SERVERS_NAME,bootstrapServersCustom);
        }

        //acks
        String acks = (String) configs.get(ProducerCons.ACKS_NAME);
        String acksCustom = customDefaultSenderConfig.getAcks();
        if(StringUtils.isBlank(acks) && StringUtils.isNotBlank(acksCustom)){
            rs.put(ProducerCons.ACKS_NAME,acksCustom);
        }

        //linger.ms
        Long lingerMs = (Long) configs.get(ProducerCons.LINGER_MS_NAME);
        Long lingerMsCustom = customDefaultSenderConfig.getLingerMs();
        if((lingerMs == null || lingerMs < 0) && (lingerMsCustom != null && lingerMsCustom >= 0)){
            rs.put(ProducerCons.LINGER_MS_NAME,lingerMsCustom);
        }

        //batch.size
        Integer batchSize = (Integer) configs.get(ProducerCons.BATCH_SIZE_NAME);
        Integer batchSizeCustom = customDefaultSenderConfig.getBatchSize();
        if((batchSize == null || batchSize <= 0) && (batchSizeCustom != null && batchSizeCustom > 0)){
            rs.put(ProducerCons.BATCH_SIZE_NAME, batchSizeCustom);
        }
        //key.serializer
        String keySerializer = (String) configs.get(ProducerCons.KEY_SERIALIZER_NAME);
        String keySerializerCustom = customDefaultSenderConfig.getKeySerialize();
        if(StringUtils.isBlank(keySerializer) && StringUtils.isNotBlank(keySerializerCustom)){
            rs.put(ProducerCons.KEY_SERIALIZER_NAME,keySerializerCustom);
        }

        //value.serializer
        String valueSerializer = (String) configs.get(ProducerCons.VALUE_SERIALIZER_NAME);
        String valueSerializerCustom = customDefaultSenderConfig.getValueSerializer();
        if(StringUtils.isBlank(valueSerializer) && StringUtils.isNotBlank(valueSerializerCustom)){
            rs.put(ProducerCons.VALUE_SERIALIZER_NAME,valueSerializerCustom);
        }

        //client.id
//        String clientId = (String) configs.get(ProducerCons.CLIENT_ID_NAME);
//        String clientIdCustom = customDefaultSenderConfig.getClientId();
//        if(StringUtils.isBlank(clientId) && StringUtils.isNotBlank(clientIdCustom)){
//            rs.put(ProducerCons.CLIENT_ID_NAME,clientIdCustom);
//        }

        //retries
        Long retries = (Long) configs.get(ProducerCons.RETRIES_NAME);
        Integer retriesCustom = customDefaultSenderConfig.getRetries();
        if((retries == null || retries < 0) && (retriesCustom != null && retriesCustom >= 0)){
            rs.put(ProducerCons.RETRIES_NAME,retriesCustom);
        }

        //buffer.memory
        Long bufferMemory = (Long) configs.get(ProducerCons.BUFFER_MEMORY_NAME);
        Long bufferMemoryCustom = customDefaultSenderConfig.getBufferMemory();
        if((bufferMemory == null || bufferMemory <= 0) && (bufferMemoryCustom != null && bufferMemoryCustom > 0)){
            rs.put(ProducerCons.BUFFER_MEMORY_NAME,bufferMemoryCustom);
        }

        return rs;
    }

    /**
     * 全局统一默认配置
     * @return
     */
    private Properties populateCommonDefaultConfigs(Properties configs){
        Properties rs = new Properties();
        rs.putAll(configs);
        //broker 地址列表
        String bootstrapServers = (String)configs.get(ProducerCons.BOOTSTRAP_SERVERS_NAME);
        if(StringUtils.isBlank(bootstrapServers) && StringUtils.isNotBlank(producerDefaultConfig.getBootstrapServers())){
            rs.put(ProducerCons.BOOTSTRAP_SERVERS_NAME, producerDefaultConfig.getBootstrapServers());
        }

        //acks
        String acks = (String) configs.get(ProducerCons.ACKS_NAME);
        if(StringUtils.isBlank(acks) && StringUtils.isNotBlank(producerDefaultConfig.getAcks())){
            rs.put(ProducerCons.ACKS_NAME,producerDefaultConfig.getAcks());
        }

        //linger.ms
        Long lingerMs = (Long) configs.get(ProducerCons.LINGER_MS_NAME);
        if((lingerMs == null || lingerMs < 0) && (producerDefaultConfig.getLingerMs() != null && producerDefaultConfig.getLingerMs() >= 0)){
            rs.put(ProducerCons.LINGER_MS_NAME,producerDefaultConfig.getLingerMs());
        }

        //batch.size
        Long batchSize = (Long) configs.get(ProducerCons.BATCH_SIZE_NAME);
        if((batchSize == null || batchSize <= 0) && (producerDefaultConfig.getBatchSize() != null && producerDefaultConfig.getBatchSize() > 0)){
            rs.put(ProducerCons.BATCH_SIZE_NAME,producerDefaultConfig.getBatchSize());
        }
        //key.serializer
        String keySerializer = (String) configs.get(ProducerCons.KEY_SERIALIZER_NAME);
        if(StringUtils.isBlank(keySerializer) && StringUtils.isNotBlank(producerDefaultConfig.getKeySerializer())){
            rs.put(ProducerCons.KEY_SERIALIZER_NAME,producerDefaultConfig.getKeySerializer());
        }

        //value.serializer
        String valueSerializer = (String) configs.get(ProducerCons.VALUE_SERIALIZER_NAME);
        if(StringUtils.isBlank(valueSerializer) && StringUtils.isNotBlank(producerDefaultConfig.getValueSerializer())){
            rs.put(ProducerCons.VALUE_SERIALIZER_NAME,producerDefaultConfig.getValueSerializer());
        }

        //client.id
//        String clientId = (String) configs.get(ProducerCons.CLIENT_ID_NAME);
//        if(StringUtils.isBlank(clientId) && StringUtils.isNotBlank(producerDefaultConfig.getClientId())){
//            rs.put(ProducerCons.CLIENT_ID_NAME,producerDefaultConfig.getClientId());
//        }

        //retries
        Long retries = (Long) configs.get(ProducerCons.RETRIES_NAME);
        if((retries == null || retries < 0) && (producerDefaultConfig.getRetries() != null && producerDefaultConfig.getRetries() >= 0)){
            rs.put(ProducerCons.RETRIES_NAME,producerDefaultConfig.getRetries());
        }

        //buffer.memory
        Long bufferMemory = (Long) configs.get(ProducerCons.BUFFER_MEMORY_NAME);
        if((bufferMemory == null || bufferMemory <= 0) && (producerDefaultConfig.getBufferMemory() != null && producerDefaultConfig.getBufferMemory() > 0)){
            rs.put(ProducerCons.BUFFER_MEMORY_NAME,producerDefaultConfig.getBufferMemory());
        }
        return rs;
    }
}
